package com.atguigu.gmall.realtime.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.java.tuple.Tuple2;
import redis.clients.jedis.Jedis;

import java.util.List;

/**
 * Author: Felix
 * Date: 2022/1/21
 * Desc: 查询维度的工具类
 */
public class DimUtil {
    public static JSONObject getDimInfo(String tableName, String id) {
        return getDimInfo(tableName,Tuple2.of("id",id));
    }
    /**
     * 查询维度数据  加入旁路缓存进行优化
     * 思路:先从缓存中查询维度数据，如果缓存中存在维度数据，直接将维度数据返回（缓存命中）；
     *  如果缓存中不存在当前查询的维度数据，那么发送请求到Phoenix表中查询，并将查询的结果
     *  放到缓存中
     * 选型：状态   Redis(√)
     * Redis分析：     类型：String       TTL:1day    key:    dim:dim_base_trademark:14_aa_bb
     * @param tableName
     * @param columnAndValues
     * @return
     */
    public static JSONObject getDimInfo(String tableName, Tuple2<String, String>... columnAndValues) {
        //拼接查询Redis的key
        StringBuilder redisKey = new StringBuilder("dim:"+tableName.toLowerCase()+":");
        //拼接查询SQL
        StringBuilder selectSql = new StringBuilder("select * from " + tableName + " where ");
        //对可变长参数进行遍历
        for (int i = 0; i < columnAndValues.length; i++) {
            Tuple2<String, String> columnAndValue = columnAndValues[i];
            String columnName = columnAndValue.f0;
            String columnValue = columnAndValue.f1;
            selectSql.append(columnName + "='" + columnValue + "'");
            redisKey.append(columnValue);
            if(i < columnAndValues.length - 1){
                selectSql.append(" and ");
                redisKey.append("_");
            }
        }

        Jedis jedis = null;
        String jsonStr = null;
        JSONObject dimJsonObj = null;

        //从Redis中查询维度数据
        try {
            jedis = RedisUtil.getJedis();
            jsonStr = jedis.get(redisKey.toString());
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("从Redis中查询维度数据发送了异常");
        }

        //判断是否命中缓存
        if(jsonStr != null && jsonStr.length() > 0){
            //命中
            dimJsonObj = JSON.parseObject(jsonStr);
        }else{
            //在Redis中没有找到维度数据，发送请求到Phoenix中去查询维度
            System.out.println("从Phoenix表中查询维度的SQL:" + selectSql);
            //从Phoenix表中查询数据，底层还是调用PhoenixUtil中的方法
            List<JSONObject> jsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);

            if(jsonObjList != null && jsonObjList.size() > 0){
                //从Phoenix表中查询 到了维度数据
                dimJsonObj = jsonObjList.get(0);
                //将查询到的维度数据放到Redis中缓存起来
                if(jedis != null){
                    jedis.setex(redisKey.toString(),3600*24,dimJsonObj.toJSONString());
                }
            }else{
                //从Phoenix表中没有查询到维度数据
                System.out.println("在维度表中没有该维度数据" + selectSql);
            }
        }

        //关闭Redis客户端
        if(jedis != null){
            System.out.println("---关闭Redis客户端---");
            jedis.close();
        }

        return dimJsonObj;
    }

    //根据维度主键，获取维度对象   select * from dim_base_trademark where id='14' and tm_name='aa'
    public static JSONObject getDimInfoNoCache(String tableName, Tuple2<String, String>... columnAndValues) {
        //拼接查询SQL
        StringBuilder selectSql = new StringBuilder("select * from " + tableName + " where ");
        //对可变长参数进行遍历
        for (int i = 0; i < columnAndValues.length; i++) {
            Tuple2<String, String> columnAndValue = columnAndValues[i];
            String columnName = columnAndValue.f0;
            String columnValue = columnAndValue.f1;
            selectSql.append(columnName + "='" + columnValue + "'");
            if(i < columnAndValues.length - 1){
                selectSql.append(" and ");
            }
        }
        System.out.println("从Phoenix表中查询维度的SQL:" + selectSql);

        //从Phoenix表中查询数据，底层还是调用PhoenixUtil中的方法
        List<JSONObject> jsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
        JSONObject dimJsonObj = null;
        if(jsonObjList != null && jsonObjList.size() > 0){
            //从Phoenix表中查询 到了维度数据
            dimJsonObj = jsonObjList.get(0);
        }else{
            //从Phoenix表中没有查询到维度数据
            System.out.println("在维度表中没有该维度数据" + selectSql);
        }
        return dimJsonObj;
    }

    //清除Redis中的缓存数据
    public static void deleteCached(String tableName,String id){
        String redisKey = "dim:"+tableName.toLowerCase()+":" + id;
        try {
            Jedis jedis = RedisUtil.getJedis();
            jedis.del(redisKey);
            jedis.close();
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("从Redis中闪出去缓存数据失败~~~");
        }
    }

    public static void main(String[] args) {
        //JSONObject dimInfo = getDimInfoNoCache("dim_base_trademark",
        //    Tuple2.of("id", "14"),Tuple2.of("tm_name","aa"));
        //JSONObject dimInfo = getDimInfo("dim_base_trademark",Tuple2.of("id", "14"));
        JSONObject dimInfo = getDimInfo("dim_base_trademark","17");
        System.out.println(dimInfo);
    }
}
